package com.my.test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.json.JSONObject
import java.util.Date
import com.saic.utils.HttpSaicUtils
import java.text.SimpleDateFormat

/**
 * @author ZhiLi
 */
object Hello {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val url = "jdbc:mysql://localhost:3306/mine"
    val jdbcDF = sqlContext.read.format("jdbc").options(
          Map("url" -> url, 
              "dbtable" -> "hbasekey", 
              "user" -> "root", 
              "password" -> "admin", 
              "driver" -> "com.mysql.jdbc.Driver")
              ).load()
     val db_name = "mine"
     val table_name = "hbasekey"
     val schema = jdbcDF.schema
     val tableRDD = jdbcDF.where("1=1")
     tableRDD.registerTempTable("data_table_for_spark")
     sqlContext.cacheTable("data_table_for_spark")
//     var colDF = jdbcDF.select(schema.fields.apply(1).name)
     val tableCount = tableRDD.count()
    
   var distinctCount:Long = 0
    var nullCount:Long = 0
    var nullStringCount:Long = 0
    var notNullCount:Long = 0
    var maxValue = ""
    var minValue = ""
    var maxLength = 0
    var minLength = 0
    val indexDateFormat = new SimpleDateFormat("-yyyyMM");
    val timeFomat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val yyyymm = indexDateFormat.format(new Date)
    val date = System.currentTimeMillis()
    
    
    val detailsUrl = "http://192.168.40.130:9200/table-count"+yyyymm+"/details"
    val basicUrl = "http://192.168.40.130:9200/table-count"+yyyymm+"/basic"
    val detailsJson:JSONObject = new JSONObject()
    val basicNotNullJson:JSONObject = new JSONObject()
    val basicNullJson:JSONObject = new JSONObject()
    val basicNullStrJson:JSONObject = new JSONObject()
    
    schema.foreach { x =>
      
      detailsJson.append("db_name", db_name)
      detailsJson.append("table_name", table_name)
      detailsJson.append("column_name", x.name.mkString)
      detailsJson.append("column_type", x.dataType.typeName)
      detailsJson.append("timestamp", date)
      
      
      val colDF = jdbcDF.select(x.name)
      distinctCount = colDF.distinct().count()
      nullCount = colDF.filter(x.name +"= null").count()
      nullStringCount = colDF.filter(x.name +"=''").count()
      notNullCount = tableCount-nullCount-nullStringCount
      colDF.describe(x.name).foreach { x =>  
        if(x.apply(0).toString().equals("max")){
          maxValue = x.apply(1)+"" 
        }else if (x.apply(0).toString().equals("min")) {
          minValue = x.apply(1)+""
        }
      }

      maxLength = colDF.map { x => (x.apply(0)+"").length }.max()
      minLength = colDF.map { x => (x.apply(0)+"").length }.min()
      
      detailsJson.append("row_count", tableCount)
      detailsJson.append("row_discount", distinctCount)
      detailsJson.append("null_count", nullCount)
      detailsJson.append("null_stringcount", nullStringCount)
      detailsJson.append("notnull_count", notNullCount)
      detailsJson.append("max_value", maxValue)
      detailsJson.append("min_value", minValue)
      detailsJson.append("max_length", maxLength)
      detailsJson.append("min_length", minValue)
      
      
      basicNotNullJson.append("db_name", db_name)
      basicNotNullJson.append("table_name", table_name)
      basicNotNullJson.append("column_name", x.name.mkString)
      basicNotNullJson.append("column_type", x.dataType.typeName)
      basicNotNullJson.append("timestamp", date)
      basicNotNullJson.append("string_type", "not_null_count")
      basicNotNullJson.append("row_count", notNullCount)
      
      basicNullJson.append("db_name", db_name)
      basicNullJson.append("table_name", table_name)
      basicNullJson.append("column_name", x.name.mkString)
      basicNullJson.append("column_type", x.dataType.typeName)
      basicNullJson.append("timestamp", date)
      basicNullJson.append("string_type", "null_count")
      basicNullJson.append("row_count", nullCount)
      
      basicNullStrJson.append("db_name", db_name)
      basicNullStrJson.append("table_name", table_name)
      basicNullStrJson.append("column_name", x.name.mkString)
      basicNullStrJson.append("column_type", x.dataType.typeName)
      basicNullStrJson.append("timestamp", date)
      basicNullStrJson.append("string_type", "null_string_count")
      basicNullStrJson.append("row_count", nullStringCount)
      
      HttpSaicUtils.doPost(basicUrl, basicNotNullJson)
      HttpSaicUtils.doPost(basicUrl, basicNullJson)
      HttpSaicUtils.doPost(basicUrl, basicNullStrJson)
      
      HttpSaicUtils.doPost(detailsUrl, detailsJson)
//      HttpSaicUtils.doPost("http://192.168.40.130:9200/mine5/hbasekey", json)
      }
    
     sc.stop()
  }
}